/**
 * Copyright 2014-2018 yangming.liu<bytefox@126.com>.
 * <p>
 * This copyrighted material is made available to anyone wishing to use, modify,
 * copy, or redistribute it subject to the terms and conditions of the GNU
 * Lesser General Public License, as published by the Free Software Foundation.
 * <p>
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
 * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License
 * for more details.
 * <p>
 * You should have received a copy of the GNU Lesser General Public License
 * along with this distribution; if not, see <http://www.gnu.org/licenses/>.
 */
package org.feisoft.common.utils;

import com.caucho.hessian.io.HessianInput;
import com.caucho.hessian.io.HessianOutput;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.pool.KryoCallback;
import com.esotericsoftware.kryo.pool.KryoFactory;
import com.esotericsoftware.kryo.pool.KryoPool;
import org.nustaq.serialization.FSTConfiguration;
import org.objenesis.strategy.InstantiatorStrategy;
import org.objenesis.strategy.StdInstantiatorStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;

public class SerializeUtils
{
    static final Logger logger = LoggerFactory.getLogger(SerializeUtils.class);

    static final int SERIALIZER_DEFAULT = 0x0;
    static final int SERIALIZER_KRYO = 0x1;
    static final int SERIALIZER_HESSIAN = 0x2;
    static final int SERIALIZER_FST = 0x3;

    static final FSTConfiguration fstConfig = FSTConfiguration.createDefaultConfiguration();

    static final KryoPool kryoPool = new KryoPool.Builder(new KryoFactory()
    {
        public Kryo create()
        {
            Kryo kryo = new Kryo();
            StdInstantiatorStrategy stdInstantiatorStrategy = new StdInstantiatorStrategy();
            InstantiatorStrategy instantiatorStrategy = new Kryo.DefaultInstantiatorStrategy(stdInstantiatorStrategy);
            kryo.setInstantiatorStrategy(instantiatorStrategy);
            return kryo;
        }
    }).softReferences().build();

    public static byte[] serializeObject(Serializable obj) throws IOException
    {
        int serializer = SERIALIZER_DEFAULT;
        byte[] dataArray = null;
        try
        {
            dataArray = kryoSerialize(obj);
            serializer = SERIALIZER_KRYO;
        } catch (Exception ex)
        {
            dataArray = javaSerialize(obj);
            serializer = SERIALIZER_KRYO;
        }

        byte[] byteArray = new byte[dataArray.length + 1];
        byteArray[0] = (byte) serializer;
        System.arraycopy(dataArray, 0, byteArray, 1, dataArray.length);

        return byteArray;
    }

    public static Serializable deserializeObject(byte[] bytes) throws IOException
    {
        if (bytes.length == 0)
        {
            throw new IllegalArgumentException();
        }

        byte[] byteArray = new byte[bytes.length - 1];
        int serializer = bytes[0];

        if (serializer == SERIALIZER_KRYO)
        {
            System.arraycopy(bytes, 1, byteArray, 0, byteArray.length);
            return kryoDeserialize(byteArray);
        } else if (serializer == SERIALIZER_HESSIAN)
        {
            System.arraycopy(bytes, 1, byteArray, 0, byteArray.length);
            return hessianDeserialize(byteArray);
        } else if (serializer == SERIALIZER_FST)
        {
            System.arraycopy(bytes, 1, byteArray, 0, byteArray.length);
            return fstDeserialize(byteArray);
        } else if (serializer == SERIALIZER_DEFAULT)
        {
            System.arraycopy(bytes, 1, byteArray, 0, byteArray.length);
            return javaDeserialize(byteArray);
        } else
        {
            throw new IllegalArgumentException();
        }
    }

    public static byte[] fstSerialize(final Serializable obj) throws IOException
    {
        return fstConfig.asByteArray(obj);
    }

    public static Serializable fstDeserialize(byte[] byteArray) throws IOException
    {
        return (Serializable) fstConfig.asObject(byteArray);
    }

    public static byte[] javaSerialize(final Serializable obj) throws IOException
    {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(new BufferedOutputStream(baos));
        try
        {
            oos.writeObject(obj);
        } finally
        {
            CommonUtils.closeQuietly(oos);
        }
        return baos.toByteArray();
    }

    public static Serializable javaDeserialize(byte[] byteArray) throws IOException
    {
        ObjectInputStream ois = new ObjectInputStream(new BufferedInputStream(new ByteArrayInputStream(byteArray)));
        try
        {
            return (Serializable) ois.readObject();
        } catch (ClassNotFoundException ex)
        {
            throw new IllegalStateException(ex);
        } finally
        {
            CommonUtils.closeQuietly(ois);
        }
    }

    public static byte[] kryoSerialize(final Serializable obj) throws IOException
    {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        final Output output = new Output(baos);

        try
        {
            kryoPool.run(new KryoCallback<Object>()
            {
                public Object execute(Kryo kryo)
                {
                    kryo.writeClassAndObject(output, obj);
                    return null;
                }
            });
        } finally
        {
            CommonUtils.closeQuietly(output);
        }

        return baos.toByteArray();
    }

    public static Serializable kryoDeserialize(byte[] byteArray) throws IOException
    {
        final Input input = new Input(new ByteArrayInputStream(byteArray));

        try
        {
            return kryoPool.run(new KryoCallback<Serializable>()
            {
                public Serializable execute(Kryo kryo)
                {
                    return (Serializable) kryo.readClassAndObject(input);
                }
            });
        } finally
        {
            CommonUtils.closeQuietly(input);
        }
    }

    public static byte[] hessianSerialize(Serializable obj) throws IOException
    {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        HessianOutput ho = new HessianOutput(baos);
        try
        {
            ho.writeObject(obj);
            return baos.toByteArray();
        } finally
        {
            CommonUtils.closeQuietly(baos);
        }

    }

    public static Serializable hessianDeserialize(byte[] bytes) throws IOException
    {
        ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
        HessianInput hi = new HessianInput(bais);
        try
        {
            Object result = hi.readObject();
            return (Serializable) result;
        } finally
        {
            CommonUtils.closeQuietly(bais);
        }
    }

}
